package com.eapi.project.controller;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;

import com.eapi.eapicommom.model.entity.InterfaceInfo;
import com.eapi.eapicommom.model.entity.User;
import com.eapi.project.model.entity.Invokerecord;
import com.eapi.project.service.InterfaceInfoService;
import com.eapi.project.service.InvokerecordService;
import com.eapi.project.service.UserService;
import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.params.SetParams;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;

import java.util.ArrayList;
import java.util.Collections;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@RestController
@CrossOrigin(origins = "http://localhost:8000", methods = {RequestMethod.GET}, allowedHeaders = "*", allowCredentials = "true")
public class SseSentController {

    @Resource
    UserService userService;

    @Resource
    InterfaceInfoService interfaceInfoService;
    @Resource
    InvokerecordService invokerecordService;
    private ExecutorService nonBlockingService = Executors.newCachedThreadPool();

    private AtomicBoolean isEmitterCompleted = new AtomicBoolean(false);
    private static final String HOST = "r-bp1yjtnfoml2vk5nrspd.redis.rds.aliyuncs.com";
    private static final int PORT = 6379;

    private static final String AUTH = "Zhang031101";

    private static volatile JedisPool jedisPool = null;

    private Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    static {
        jedisPool = getJedisPoolInstance();
    }

    /**
     * 获取RedisPool实例（单例）
     *
     * @return RedisPool实例
     */
    public static JedisPool getJedisPoolInstance() {
        if (jedisPool == null) {
            synchronized (SseSentController.class) {
                if (jedisPool == null) {
                    JedisPoolConfig poolConfig = new JedisPoolConfig();
                    poolConfig.setMaxTotal(1000);           // 最大连接数
                    poolConfig.setMaxIdle(32);              // 最大空闲连接数
                    poolConfig.setMaxWaitMillis(100 * 10000);  // 最大等待时间
                    poolConfig.setTestOnBorrow(true);       // 检查连接可用性, 确保获取的redis实例可用
                    jedisPool = new JedisPool(poolConfig, HOST, PORT, 10000000, AUTH);
                }
            }
        }

        return jedisPool;
    }

    Jedis jedis = null;

    @GetMapping("/ssSent")
    public SseEmitter getSseStream(@RequestParam Long interfaceInfoId, HttpServletRequest request) throws InterruptedException, ClientException {
        String name = "queue";
        String topicid = null;
        try {
            //jedisPool建议封装成工具类；并采用单例模式（桌静态对象）
            jedis = jedisPool.getResource();  // 获取Redus连接
            String jedisAuth = jedis.auth(AUTH);
            if (!jedis.exists(name)) {
                topicid = "" + 0;
                jedis.set("queue", "0", new SetParams().px(600000L));
            } else {
                String rebNumbs = jedis.get(name);
                int i = (Integer.parseInt(rebNumbs) + 1) % 10;
                topicid = "" + i;
                Long del = jedis.del(name);
                jedis.set("queue", topicid, new SetParams().px(600000L));
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (jedis != null) {
                //jedisPool.returnResourceObject(jedis);  // 已废弃，推荐使用jedis.close()方法
                jedis.close(); // 关闭redis连接,还回redis连接池中；
            }
        }
        List<SseEmitter> list = new ArrayList();
        for (int i = 0; i < 10; i++) {
            SseEmitter emitter = new SseEmitter(120000L);
            list.add(emitter);
        }

        // 获取用户信息
        User loginUser = userService.getById(7);
        Long userId = loginUser.getId();

        // 获取接口信息
        InterfaceInfo interfaceInfo = interfaceInfoService.getById(interfaceInfoId);
        String interfaceInfoName = interfaceInfo.getName();
        String description = interfaceInfo.getDescription();

        // 消息队列相关配置
        String endpoints = "rmq-cn-0mm3r8c4q05.cn-hangzhou.rmq.aliyuncs.com:8080";
        String topic = "topic" + topicid;
        String consumerGroup = "chat" + topicid;
        int num = Integer.parseInt(topicid);
        System.out.println("consumerGroup----------------" + consumerGroup);
        System.out.println("topic----------------" + topic);
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);


        builder.setCredentialProvider(new StaticSessionCredentialsProvider("92cy6p7KnA4M29DS", "pvV1mi5M46Gvn39S"));
        ClientConfiguration clientConfiguration = builder.build();
        //订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        System.out.println("订阅成功");
        //初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        // 订阅另外一个Topic，如需取消订阅该Topic，请删除该部分的订阅代码，重新启动消费端即可。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                //设置消费者分组。
                .setConsumerGroup(consumerGroup)
                //设置预绑定的订阅关系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                //设置消费监听器。‘
                .setMessageListener(messageView -> {
                    synchronized (messageView) {
                        //处理消息并返回消费结果。
                        String bodyStr = messageView.getProperties().toString();
                        System.out.println("bodyStr----------------" + bodyStr);
                        try {
                            Thread.sleep(500);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        Pattern pattern = Pattern.compile("\"(.*?)\"");
                        Matcher matcher = pattern.matcher(bodyStr);
                        //System.out.println(bodyStr);
                        StringBuilder cleanedText = new StringBuilder();
                        try {
                            if (matcher.find()) {
                                String group = matcher.group(1);
                                cleanedText.append(group);
                                list.get(num).send(group);
                            } else {
                                int length = cleanedText.length();
                                double price = length * 0.012;
                                // 实际扣费
                                Long l = price > 1 ? (long) price : 1;
                                Long points = (loginUser.getRemainingPoints() - l);
                                loginUser.setRemainingPoints(points);
                                // 更新账户余额
                                userService.updateById(loginUser);
                                // 更新调用记录
                                Invokerecord invokerecord = new Invokerecord();
                                invokerecord.setInterfaceInfoId(interfaceInfoId);
                                invokerecord.setUserId(userId);
                                invokerecord.setInvokeCosts(l);
                                invokerecord.setInterfaceInfoName(interfaceInfoName);
                                invokerecord.setDescription(description);
                                invokerecordService.insertInvoke(invokerecord);
                                if (bodyStr.contains("E-API-endpoints")) {
                                    list.get(num).complete();
                                }
                            }
                        } catch (Exception ex) {
                            System.out.println(ex);
                            System.out.println("发生异常");
                            list.get(num).completeWithError(ex);
                        }
                    }
                    return ConsumeResult.SUCCESS;
                })
                .build();

        return list.get(num);
    }

    public String getContent(String bodyStr, String tag) {
        String contentMessage = null;
        try {
            JSONObject fullResponse = JSONUtil.parseObj(bodyStr);
            contentMessage = fullResponse.getStr(tag);
        } catch (Exception e) {
            System.out.println("Error: " + e);
        }

        return contentMessage;
    }


}
